package cn.me.alphamq.broker.netty.nameserver;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.config.BrokerProperties;
import cn.me.alphamq.broker.config.ClusterProperties;
import cn.me.alphamq.broker.event.model.nameserver.ChangeMasterEvent;
import cn.me.alphamq.broker.event.model.nameserver.Shift2MasterEvent;
import cn.me.alphamq.broker.event.spi.listener.nameserver.ChangeMasterListener;
import cn.me.alphamq.broker.event.spi.listener.nameserver.Shift2MasterListener;
import cn.me.alphamq.common.entity.req.ChangeMasterReq;
import cn.me.alphamq.common.entity.req.ServiceRegistryReq;
import cn.me.alphamq.common.entity.resp.PullBrokerAddressResp;
import cn.me.alphamq.common.enumeration.BrokerRoleEnum;
import cn.me.alphamq.common.enumeration.ServiceInstanceTypeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.remote.SyncFuture;
import cn.me.alphamq.common.util.AttrUtil;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.Inet4Address;
import java.util.HashMap;
import java.util.Map;

/**
 * broker 针对 nameserver 的处理器，可以被所有线程共享
 *
 * @author f
 */
@ChannelHandler.Sharable
public class BrokerNameServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger log = LoggerFactory.getLogger(BrokerNameServerHandler.class);

    private static final Shift2MasterListener SHIFT_2_MASTER_LISTENER = new Shift2MasterListener();
    private static final ChangeMasterListener CHANGE_MASTER_LISTENER = new ChangeMasterListener();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();

        if (EventCodeEnum.ERROR_ACCOUNT_RESP.getCode() == code) {
            // 注册失败
            throw new RuntimeException("error username or password");
        } else if (EventCodeEnum.REGISTRY_SUCCESS_RESP.getCode() == code) {
            // 注册成功，开启一个定时任务，上报心跳数据给 nameserver
            Container.getHeartBeatTaskRunner().startTask();
            log.info("register to nameserver...");
        } else if (EventCodeEnum.UN_REGISTRY_RESP.getCode() == code) {
            log.info("un registry"); // TODO DELETE
        } else if (EventCodeEnum.EMPTY_IDENTIFY_RESP.getCode() == code) {
            throw new RuntimeException("empty identify"); // TODO DELETE
        } else if (EventCodeEnum.HEART_BEAT_SUCCESS_RESP.getCode() == code) {
//            log.info("心跳成功"); // TODO DELETE
        } else if (EventCodeEnum.PULL_BROKER_ADDRESS_RESP.getCode() == code) {
            SyncFuture.removeAndSetResp(PullBrokerAddressResp.class, tcpMsg);
        } else if (EventCodeEnum.SHIFT_TO_MASTER_REQ.getCode() == code) {
            SHIFT_2_MASTER_LISTENER.onReceive(new Shift2MasterEvent(ctx));
        } else if (EventCodeEnum.CHANGE_MASTER_REQ.getCode() == code) {
            CHANGE_MASTER_LISTENER.onReceive(new ChangeMasterEvent(ctx,
                    JSON.parseObject(tcpMsg.getBody(), ChangeMasterReq.class).getNewMasterAddress()));
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        try {
            // 在通道建立成功后，给 nameserver 发送注册消息
            BrokerProperties brokerProperties = Container.getBrokerProperties();
            ClusterProperties clusterProperties = Container.getClusterProperties();

            ServiceRegistryReq registryReq = new ServiceRegistryReq();
            registryReq.setUsername(brokerProperties.getNameServerUsername());
            registryReq.setPassword(brokerProperties.getNameServerPassword());
            registryReq.setIp(Inet4Address.getLocalHost().getHostName());
            registryReq.setPort(brokerProperties.getBrokerPort());
            registryReq.setServiceInstanceType(ServiceInstanceTypeEnum.BROKER.getCode());

            BrokerRoleEnum brokerRole = clusterProperties.getBrokerRole();
            Map<String, Object> attrs = new HashMap<>();
            AttrUtil.putRoleToAttrs(attrs, brokerRole.getCode());
            if (BrokerRoleEnum.SINGLE != brokerRole) {
                AttrUtil.putGroupToAttrs(attrs, clusterProperties.getBrokerGroup());
            }
            registryReq.setAttrs(attrs);

            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.REGISTRY_REQ, JSON.toJSONBytes(registryReq)));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channel is inactive...");
    }
}